package service.temp.job.matching.poc.listingapi;

import io.github.resilience4j.ratelimiter.RateLimiterConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.codehaus.jackson.type.TypeReference;
import service.temp.job.matching.poc.listingapi.dto.ListResult;
import service.temp.job.matching.poc.listingapi.dto.SearchItemDetail;
import service.temp.job.matching.poc.listingapi.feign.FeignClientApiBuilder;
import utils.JsonUtil;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class SearchApiQueryThousand {


    public static SearchApi searchApi;

    private static final int EXECUTION_TIME_OUT_MS = 30_000;

    static {
        searchApi = buildSearchApi("22dcfb0e-1804-4df7-b8bb-15c707cbcace");
    }


    public static void main(String[] args) {
        run();
    }

    public static void run() {
        SearchApiQueryThousand runInstance = new SearchApiQueryThousand();
        List<String> lineList = runInstance.loadCandidateSearchKeyWordList();
        System.out.println("candidateSearchKeyWordList size : " + lineList.size());
        runInstance.concurrentCallSearchAPI(lineList);

    }

    public static SearchApi buildSearchApi(String consumerToken) {
        Function<Exception, SearchApi> fallbackFactory = cause -> condition -> {
            System.out.println("call listing search api failed, cause : " + cause.toString());
            return new ListResult();
        };

        RateLimiterConfig rateLimiterConfig =
                RateLimiterConfig.custom()
                        .timeoutDuration(Duration.ofSeconds(60))  // timeout to fetch permission
                        .limitRefreshPeriod(Duration.ofSeconds(6))  // permissions limit refresh period
                        .limitForPeriod(1500)  // permissions limit for period
                        .build();

        return FeignClientApiBuilder.buildFeignClientApi(SearchApi.class,
                consumerToken, EXECUTION_TIME_OUT_MS, rateLimiterConfig, fallbackFactory);
    }

    public static List<String> loadCandidateSearchKeyWordList() {
        String fileName = "/Users/zhfan/Downloads/pbs_keyword_list.txt";
        BufferedReader br = null;
        String line = "";
        List<String> candidateList = new ArrayList<>();

        try {
            br = new BufferedReader(new FileReader(fileName));
            while ((line = br.readLine()) != null) {
                if (StringUtils.isBlank(line) || line.length() <= 1) continue;
                candidateList.add(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        return candidateList;
    }


    /**
     * 并行调用 API 类
     */
    public class DirectRunnable implements Runnable {
        private List<String> jobList;
        private int jobId = 0;

        public DirectRunnable(List<String> jobList, int jobId) {
            this.jobList = jobList;
            this.jobId = jobId;
        }

        @Override
        public void run() {
            Map<String, List<SearchItemDetail>> queryMemo = new HashMap<>();
            System.out.println("jobId : " + jobId + " start to run.");

            for (String keyword : jobList) {
                if (!queryMemo.containsKey(keyword)) {
                    List<SearchItemDetail> searchList = searchApi.searchItemByPaging(keyword);
                    if (CollectionUtils.isNotEmpty(searchList)) {
                        queryMemo.put(keyword, searchList);
                        System.out.println(keyword + " -- searchItemByPaging result size : " + searchList.size());
                    } else {
                        System.out.println(keyword + " -- searchItemByPaging result has got empty itemList.");
                    }

                    if (queryMemo.size() % 10 == 0) {
                        cacheToLocal("/Users/zhfan/Downloads/20220718_01_thousand_search_result_" + jobId + ".txt", queryMemo);
                    }
                }
            }

            // finally store whole result file to local
            cacheToLocal("/Users/zhfan/Downloads/20220718_01_thousand_search_result_" + jobId + ".txt", queryMemo);
            System.out.println("jobId : " + jobId + " had finish run, final return size : " + queryMemo.size());
        }
    }


    public static void cacheToLocal(String fileName, Map<String, List<SearchItemDetail>> queryMemo) {
        try {
            BufferedWriter out = new BufferedWriter(new FileWriter(fileName));
            out.write(JsonUtil.writeValueAsString(queryMemo));
            out.close();
            System.out.println(fileName + " storeSearchResultToLocal success！");
        } catch (Exception e) {
            System.err.println(fileName + " storeSearchResultToLocal get error, cause : " + e);
        }
    }

    /**
     * 多线程版本
     *
     * @param candidateSearchKeyWordList
     */
    public void concurrentCallSearchAPI(List<String> candidateSearchKeyWordList) {
        Map<String, List<SearchItemDetail>> wholeCacheMap = loadAllMapFromLocal();
        List<String> filteredKeywordList = candidateSearchKeyWordList.stream()
                .filter(it -> !wholeCacheMap.containsKey(it)).collect(Collectors.toList());

        int SPLIT_JOB_NUM = 24, eachBatchSize = filteredKeywordList.size() / SPLIT_JOB_NUM;
        for (int i = 0; i < SPLIT_JOB_NUM; ++i) {
            int startPos = eachBatchSize * i, endPos = (i == SPLIT_JOB_NUM - 1) ? filteredKeywordList.size() : eachBatchSize * (i + 1);
            new Thread(new DirectRunnable(filteredKeywordList.subList(startPos, endPos), i)).start();
        }
    }

    public static Map<String, List<SearchItemDetail>> loadAllMapFromLocal() {
        Map<String, List<SearchItemDetail>> wholeMap = new HashMap<>();
        String[] dirList = new File("/Users/zhfan/Documents/cache01").list();

        if (dirList != null) {
            for (int i = 0; i < dirList.length; i++) {
                String fileName = "/Users/zhfan/Documents/cache01/" + dirList[i];
                Map<String, List<SearchItemDetail>> partMap = loadSearchCacheMapFromLocal(fileName);
                System.out.println("partMap size : " + partMap.size());
                wholeMap.putAll(partMap);
            }
        }

        return wholeMap;
    }

    public static Map<String, List<SearchItemDetail>> loadSearchCacheMapFromLocal(String fileName) {
        Map<String, List<SearchItemDetail>> searchMapping = new HashMap<>();
        searchMapping = JsonUtil.readValue(new File(fileName), new TypeReference<Map<String, List<SearchItemDetail>>>() {
        });

        return searchMapping;
    }

}
